package com.eshipenet.shipnet.equipmentdatahub

import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.async.ResultFuture
import org.joda.time.DateTime

import scala.concurrent.duration._
import scala.language.postfixOps

object Main {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.registerTypeWithKryoSerializer(classOf[DateTime], classOf[DateTimeSerializer])

//    val jsValueStream = createEquipmentStream(env)
    val jsValueStream = EquipmentStreamProvider.createEquipmentStream(env)
    val equipmentStream = jsValueStream
      .map(jsValue => DebeziumParser(jsValue))
      .filter(deb => deb.isSuccess)
      .map(_.get)
      .filter(_.isDefined)
      .map(_.get)
      .map(action => EquipmentAction(action))
      .map(v => { println(v); v })
      .filter(_.isSuccess)
      .map(_.get)
    equipmentStream.print("equipmentStream")

    val be = equipmentStream
      .keyBy(_.id)
      .mapWithState((action, state: Option[Map[String, EquipmentWithModel]]) => {
        action match {
          case EquipmentInsert(after) =>
            val newState = state.getOrElse(Map[String, EquipmentWithModel]()) ++ Map(after.id -> after)
            (newState, Some(newState))
          case EquipmentUpdate(_, after) =>
            val newState = state.getOrElse(Map[String, EquipmentWithModel]()) ++ Map(after.id -> after)
            (newState, Some(newState))
          case EquipmentDelete(before) =>
            val newState = state.getOrElse(Map[String, EquipmentWithModel]()) - before.id
            (newState, Some(newState))
        }
      })
      .flatMap(d => d)
      .map(_._2)

    be.print("be")

//    val dataJsStream = createEdgeChannelDataStream(env)
    val dataJsStream = EdgeChannelDataStreamProvider.createEdgeChannelDataStream(env)

    val desc = new MapStateDescriptor[String, EquipmentWithModel]("EquipmentWithModel",
                                                                  classOf[String],
                                                                  classOf[EquipmentWithModel])
    val edgeChannelDataStream = dataJsStream
      .map(EdgeChannelData(_))
      .filter(_.isSuccess)
      .map(_.get)
      .keyBy(_.id)
//    edgeChannelDataStream.print("edgeChannelDataStream")

//    val combined = edgeChannelDataStream
//      .connect(be.broadcast(desc))
//      .process(new CombineCoProcessor(desc))

    val data1 = edgeChannelDataStream
      .connect(be.broadcast(desc))
      .process(new EdgeChannelDataCoProcessor(desc))
    data1.print("data1")
    val data = data1
      .filter(_.isSuccess)
      .map(_.get)

    data.print("data")
    data.addSink(EquipmentDataSinkProvider.createSink)

//    val ed = combined.map { c =>
//      c._2.coerceDataType(c._1.)
//      val name = c._1.idToName(c._2.id)
//      val data = Map(
//        name -> c._2,
//      )
//      val timestamp = c._2.timestampNS
//      EquipmentData(c._1.equipment, c._1.model, data, timestamp)
//    }
//    ed.addSink(EquipmentDataSinkProvider.createSink)

//    val timeout    = 10 seconds
//    val maxTimeout = timeout + (1 seconds)
//
//    combined
//      .map(v => v) // why
//      .keyBy(_._1.model.id)
//      .map(v => IterableData(Some(v), None, None))
//      .iterate(d => {
//        val r = d.filter(_.ed.isDefined).keyBy(_.ed.get._1.model.id).process(new IterableDataProcessor)
//        val feedback =
//          AsyncDataStream.unorderedWait(r.filter(_.delay.isDefined), maxTimeout._1, maxTimeout._2, 10000) {
//            (in, fut: ResultFuture[IterableData]) =>
//              val delay  = in.delay.get
//              val d      = delay.toMillis / timeout.toMillis
//              val rDelay = if (d > 0) timeout else delay
//              odelay.Delay(rDelay) {
//                fut.complete(List(in))
//              }
//          }
//        (feedback, r.filter(_.data.isDefined))
//      })
//      .map(_.data.get)
//      .addSink(EquipmentDataSinkProvider.createSink)

//    combined.addSink(kafkaSink)
    val name = this.getClass.getName
    env.execute(name)
  }
}
